package com.feidee.fd.sml.algorithm.component.feature

import java.util.Random

import org.apache.spark.ml.PipelineStage
import org.apache.spark.ml.feature.{ChiSqSelector, StringIndexer}
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.types.NumericType

/**
  * @Author dongguosheng
  * @Date 2019/1/10 11:17
  * @Description songhaicheng
  * @Reviewer songhaicheng
  */
case class ChiSqSelectEncoderParam(
                                    override val input_pt: String,
                                    override val output_pt: String,
                                    override val hive_table: String,
                                    override val flow_time: String,
                                    // 等同于特征列
                                    override val inputCol: String,
                                    override val outputCol: String,
                                    override val preserveCols: String,
                                    override val modelPath: String,
                                    // 通过卡方检验选取最具有预测能力的 Top(num) 个特征，>= 1，默认 50
                                    numTopFeatures: Int,
                                    // 百分位数跟numTopFeatures类似，但按比例选择所有特征的一部分，而不是固定数量。取值范围 [0.0, 1.0]，默认 0.1
                                    percentile: Double,
                                    // 选择 P 值低于门限值的特征，这样就可以控制 false positive rate 来进行特征选择，取值范围 [0.0, 1.0]，算法默认 0.05
                                    fpr: Double,
                                    // 伪发现率上限值，取值范围 [0.0, 1.0]，算法默认 0.05
                                    fdr: Double,
                                    // 总的一类错误率，取值范围 [0.0, 1.0]，算法默认 0.05
                                    fwe: Double,
                                    // 卡方选择器类型，支持 [numTopFeatures, percentile, fpr, fdr, fwe]，默认 numTopFeatures
                                    selectorType: String,
                                    // 标签列
                                    labelCol: String
                                  ) extends FeatureParam {

  def this() = this(null, null, null, null, "input", "output", null, null,
    50, 0.1, 0.05, 0.05, 0.05, "numTopFeatures", "label")


  override def verify(): Unit = {
    require(tool.isNotNull(input_pt), "param input_pt can't be null")
    require(tool.isNotNull(output_pt), "param output_pt can't be null")
    require(tool.isNotNull(inputCol), "输入字段 inputCol 不能为空")
    require(tool.isNotNull(outputCol), "输出字段 outputCol 不能为空")
    require(!tool.isNotNull(preserveCols) || !preserveCols.contains(outputCol),
      "保留字段 preserveCols 不能与输出字段 outputCol 出现重合")
    require(numTopFeatures >= 1, "param numTopFeatures must be grater than 0")
    require(percentile >= 0 && percentile <= 1, "param percentile can't be negative")
    require(fpr >= 0 && fpr <= 1, "param fpr's range is [0.0, 1.0]")
    require(fdr >= 0 && fdr <= 1, "param fdr's range is [0.0, 1.0]")
    require(fwe >= 0 && fwe <= 1, "param fwe's range is [0.0, 1.0]")
    val selectorTypes = Array("numTopFeatures" , "percentile", "fpr", "fdr", "fwe")
    require(selectorTypes.map(_.toLowerCase).contains(selectorType.toLowerCase), s"param selectorType only accepts " +
      s"${selectorTypes.mkString("[", ", ", "]")}, but has $selectorType")
  }

  override def toMap: Map[String, Any] = {
    var map = super.toMap
    map += ("numTopFeatures" -> numTopFeatures)
    map += ("percentile" -> percentile)
    map += ("fpr" -> fpr)
    map += ("fdr" -> fdr)
    map += ("fwe" -> fwe)
    map += ("selectorType" -> selectorType)
    map += ("labelCol" -> labelCol)
    map
  }
}


class ChiSqSelectEncoder extends AbstractFeatureEncoder[ChiSqSelectEncoderParam] {

  override def setUp(param: ChiSqSelectEncoderParam, data: DataFrame): Array[PipelineStage] = {
    val selector = new ChiSqSelector()
      .setNumTopFeatures(param.numTopFeatures)
      .setFeaturesCol(param.inputCol)
      .setLabelCol(param.labelCol)
      .setOutputCol(param.outputCol)
      .setFdr(param.fdr)
      .setFwe(param.fwe)
      .setFpr(param.fpr)
      .setPercentile(param.percentile)
      .setSelectorType(param.selectorType)

    // 如果标签类型为非数字类型，则需要做索引化
    if (!data.schema(param.labelCol).dataType.isInstanceOf[NumericType]) {

      val strIdx = new StringIndexer()
        .setInputCol(param.labelCol)
        .setOutputCol(tool.renameDuplicatedColName(s"${param.labelCol}_${param.outputCol}_idx", data.schema.fieldNames))
//      data.drop(strIdx.getOutputCol)
      selector.setLabelCol(strIdx.getOutputCol)
      Array(strIdx, selector)
    } else {
      Array(selector)
    }
  }

}

object ChiSqSelectEncoder {

  def apply(paramStr: String): Unit = {
    new ChiSqSelectEncoder()(paramStr)
  }

  def main(args: Array[String]): Unit = {
    ChiSqSelectEncoder(args(0))
  }
}
